package com.nami.gate.router.middleware.mq.consumer;


import com.nami.gate.router.config.rocketmq.ConsumerConfig;
import com.nami.gate.router.config.rocketmq.RocketMqConfig;
import com.nami.gate.router.middleware.mq.MqConsumer;
import com.nami.gate.router.middleware.mq.MqSender;
import com.nami.gate.router.middleware.mq.consumer.factory.RocketMqConsumerFactory;
import com.nami.gate.router.middleware.mq.consumer.listener.RouterLoginListenerImpl;
import com.nami.gate.router.middleware.mq.consumer.listener.PublishListenerImpl;
import com.nami.gate.router.middleware.mq.consumer.model.ConsumerModel;
import com.nami.im.network.model.enums.MqTopicConst;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;


public class RocketMqConsumerImpl implements MqConsumer {

    private static final Logger log = LoggerFactory.getLogger(RocketMqConsumerImpl.class);

    private RocketMqConfig config;
    private MqSender sender;

    public RocketMqConsumerImpl(RocketMqConfig config, MqSender sender) {
        this.config = config;
        this.sender = sender;
    }

    private List<DefaultMQPushConsumer> consumers;

    @Override
    public void start() throws Exception {
        // 初始化mq receiver
        // 1.初始化 消费者组，topic, tags, 回调invoke
        // 2.组成list
        // 3.循环创建多个消费者订阅

        ConsumerConfig consumerConfig = config.getConsumerConfig();

        List<ConsumerModel> wrappers = new ArrayList<>();
        ConsumerModel login = new ConsumerModel();
        login.setTopic(MqTopicConst.CONNECT);
        login.setTags(config.getTags());
        login.setConsumerGroup(consumerConfig.getLoginConsumerGroup());
        login.setMessageListener(new RouterLoginListenerImpl(sender));
        login.setNamesrvAddr(config.getNamesrvAddr());
        login.setMessageModel(MessageModel.CLUSTERING);
        wrappers.add(login);

        ConsumerModel publish = new ConsumerModel();
        publish.setNamesrvAddr(config.getNamesrvAddr());
        publish.setTopic(MqTopicConst.PUBLISH);
//        publish.setTags(config.getTags());
        publish.setConsumerGroup(consumerConfig.getPublishConsumerGroup());
        publish.setMessageModel(MessageModel.CLUSTERING);
        publish.setMessageListener(new PublishListenerImpl(sender));
        wrappers.add(publish);


        consumers = new ArrayList<>(wrappers.size());
        for (ConsumerModel wrapper : wrappers) {
            DefaultMQPushConsumer consumer = RocketMqConsumerFactory.startConsumer(wrapper);
            log.info("mq consumer start success. topic={}|{}, consumerGroup:{}, mode:{}", wrapper.getTopic(), wrapper.getTags(), wrapper.getConsumerGroup(), wrapper.getMessageModel());
            consumers.add(consumer);
        }
    }

    @Override
    public void shutdown() {
        for (DefaultMQPushConsumer consumer : consumers) {
            if (consumer != null) {
                consumer.shutdown();
            }
        }
    }

}
